#include "mysql_handler.h"
#include "handlers.h"
#include <frpc/frpc.h>

#include <cppbase/logging.h>
#include <filesystem>
#include <workflow/Workflow.h>
#include <workflow/WFTaskFactory.h>
#include <workflow/MySQLResult.h>
#include <workflow/WFFacilities.h>
#include <fstream>
#include<cstdlib>
#include <cppbase/utility.h>
using namespace protocol;
#define CHECK_MYSQL_STATUS(task)                                                                                \
	if (task->get_state() != WFT_STATE_SUCCESS)                                                                 \
	{                                                                                                           \
		std::string err = WFGlobal::get_error_string(task->get_state(), task->get_error());                     \
		((Frpc::Reply *)task->user_data)->set_bad_output(Frpc::Status::InternalServerError); \
		return;                                                                                                 \
	}
namespace fs = std::filesystem;
namespace DBHandler
{
	void cache_exchanges(WFMySQLTask *task)
	{
		if (task->get_state() != WFT_STATE_SUCCESS)
		{
			std::cout << (WFGlobal::get_error_string(task->get_state(), task->get_error())) << "\n";
			return;
		}
		MySQLResponse *resp = task->get_resp();
		MySQLResultCursor cursor(resp);
		std::vector<MySQLCell> arr;
		Frpc::ExchangeArray exch;

		while (cursor.fetch_row(arr))
		{
			auto c = exch.add_exchange();
			c->set_id(arr[0].as_int());
			c->set_code(arr[1].as_string());
			c->set_name(arr[2].as_string());
			c->set_currency(arr[3].as_string());
		}
	//	DLOG("SQL ExchangeArray size:%ld", exch.ByteSizeLong());
		std::ofstream ofs;
		fs::path p(std::to_string(int(Command::GetExchanges)));
		std::error_code ec;
		int size = fs::file_size(p, ec);
		if (!ec && size == exch.ByteSizeLong())
		{
			return;
		}
		else
		{
			ofs.open(p, std::ios::out | std::ios::trunc);
			std::string buf;
			buf.resize(exch.ByteSizeLong());
			exch.SerializeToString(&buf);
			ofs.write(buf.data(), buf.size());
			ofs.close();
		}
	}
	std::string make_qury_contracts(int id)
	{
		if (id <= 0 || id > 9)
			DFATAL("id value error when make query string,%d", id);
		std::string tmp = "SELECT id,name,symbol,exchange_id,type_id FROM exchange_sh where type_id=1 union SELECT id,name,symbol,exchange_id,type_id FROM exchange_sz where type_id=1;";
		if (id == 1)
			return tmp;
		auto ai = tmp.find_first_of('1');
		auto bi = tmp.find_last_of('1');
		tmp[ai] = std::to_string(id)[0];
		tmp[bi] = std::to_string(id)[0];
		return tmp;
	}
	void cache_syscollection_content(WFMySQLTask *task)
	{
		if (task->get_state() != WFT_STATE_SUCCESS)
		{
			FATAL("read system collection content from sql failed.");
			std::cout << (WFGlobal::get_error_string(task->get_state(), task->get_error())) << "\n";
			return;
		}
		MySQLResultCursor cursor(task->get_resp());
		std::vector<MySQLCell> arr;
		Frpc::ContractArray contracta;
		while (cursor.fetch_row(arr))
		{
			auto c = contracta.add_contract();
			c->set_id(arr[0].as_int());
			c->set_name(arr[1].as_string());
			c->set_symbol(arr[2].as_string());
			c->set_exchange_id(arr[3].as_int());
			c->set_type_id(arr[4].as_int());
		}
		int* idptr=(int*)series_of(task)->get_context();
		DLOG("collection id:%d,SQL ContractArray row count:%d", *idptr, contracta.contract_size());
		if (contracta.contract_size() == 0)
		{
			DFATAL("fetched 0 contract rows?");
		}
		std::ofstream ofs;
		
		fs::path p(std::to_string(int(Command::GetCollectionContent)) + "_" + std::to_string(*idptr));
		std::error_code ec;
		int size = fs::file_size(p, ec);
		if (!ec && size == contracta.ByteSizeLong())
		{
			;
		}
		else
		{
			ofs.open(p, std::ios::out | std::ios::trunc);
			std::string buf;
			contracta.SerializeToString(&buf);
			ofs.write(buf.data(), buf.size());
			ofs.close();
		}
		if (*idptr != 4)
		{
			
		//	DLOG("create new sql task,id=%d", *id);
			auto next_task = WFTaskFactory::create_mysql_task(AppGlobal::instance()->get_config("mysql"), MYSQL_RETRY, cache_syscollection_content);
			*idptr+=1;
			next_task->get_req()->set_query(make_qury_contracts(*idptr));
			*series_of(task) << next_task;
		}
	}
	void cache_syscollections(WFMySQLTask *task)
	{
		if (task->get_state() != WFT_STATE_SUCCESS)
		{
			std::cout << (WFGlobal::get_error_string(task->get_state(), task->get_error())) << "\n";
			return;
		}
		MySQLResponse *resp = task->get_resp();
		MySQLResultCursor cursor(resp);
		std::vector<MySQLCell> arr;

		Frpc::CollectionArray colla;
		while (cursor.fetch_row(arr))
		{
			auto c = colla.add_collection();
			c->set_id(arr[0].as_int());
			c->set_name(arr[1].as_string());
			c->set_type_id(arr[2].as_int());
		}
		DLOG("SQL CollectionArray rows:%d", colla.collection_size());
		std::ofstream ofs;
		ofs.open(std::to_string(int(Command::GetSysCollections)), std::ios::out | std::ios::trunc);
		std::string buf;
		//buf.resize(colla.ByteSizeLong());
		colla.SerializeToString(&buf);
		ofs.write(buf.data(), buf.size());
		ofs.close();

		//start cache first collection content
		auto next_task = WFTaskFactory::create_mysql_task(AppGlobal::instance()->get_config("mysql"), MYSQL_RETRY, cache_syscollection_content);
		next_task->get_req()->set_query(make_qury_contracts(1));
		auto p=malloc(sizeof(int));
		*(int*)p=1;
		series_of(task)->set_context(p);
		*series_of(task) << next_task;
		task->set_callback([p](const WFMySQLTask *){DLOG("free p");free(p); });
	}
	void init_cache()
	{
		auto cache_root=AppGlobal::instance()->get_config("cache_dir");
		auto uri= AppGlobal::instance()->get_config("mysql");
		if(cache_root.size()<5)
			DFATAL("Error:cache dir not set.");

		fs::path cache_path(cache_root);
		cache_path/="mysql";
		fs::create_directories(cache_path);
		auto old_path=fs::current_path();
		fs::current_path(cache_path);
		WFFacilities::WaitGroup wait_group(2);

		auto task = WFTaskFactory::create_mysql_task(uri, MYSQL_RETRY, DBHandler::cache_exchanges);
		task->get_req()->set_query("select id,code,name,currency from exchange_info where id<10;");
		Workflow::start_series_work(task, [&wait_group](const SeriesWork *)
									{ wait_group.done(); });

		auto task2 = WFTaskFactory::create_mysql_task(uri, MYSQL_RETRY, DBHandler::cache_syscollections);
		task2->get_req()->set_query("select id,name,type_id from collection_info where id<10 and type_id=1;");
		Workflow::start_series_work(task2, [&wait_group](const SeriesWork *)
									{ wait_group.done(); });

		wait_group.wait();
		DLOG("exchange list and system collection content have been cached. \nDir:%s",cache_path.c_str());
		fs::current_path(old_path);
	}
	void print_result(WFMySQLTask *task)
	{
		MySQLResponse *resp = task->get_resp();

		MySQLResultCursor cursor(resp);
		const MySQLField *const *fields;
		std::vector<MySQLCell> arr;

		CHECK_MYSQL_STATUS(task);

		do
		{
			if (cursor.get_cursor_status() != MYSQL_STATUS_GET_RESULT &&
				cursor.get_cursor_status() != MYSQL_STATUS_OK)
			{
				break;
			}

			fprintf(stderr, "---------------- RESULT SET ----------------\n");

			if (cursor.get_cursor_status() == MYSQL_STATUS_GET_RESULT)
			{
				fprintf(stderr, "cursor_status=%d field_count=%u rows_count=%u\n",
						cursor.get_cursor_status(), cursor.get_field_count(),
						cursor.get_rows_count());

				//nocopy api
				fields = cursor.fetch_fields();
				for (int i = 0; i < cursor.get_field_count(); i++)
				{
					if (i == 0)
					{
						fprintf(stderr, "db=%s table=%s\n",
								fields[i]->get_db().c_str(), fields[i]->get_table().c_str());
						fprintf(stderr, "  ---------- COLUMNS ----------\n");
					}
					fprintf(stderr, "  name[%s] type[%s]\n",
							fields[i]->get_name().c_str(),
							datatype2str(fields[i]->get_data_type()));
				}
				fprintf(stderr, "  _________ COLUMNS END _________\n\n");

				while (cursor.fetch_row(arr))
				{
					fprintf(stderr, "  ------------ ROW ------------\n");
					for (size_t i = 0; i < arr.size(); i++)
					{
						fprintf(stderr, "  [%s][%s]", fields[i]->get_name().c_str(),
								datatype2str(arr[i].get_data_type()));
						if (arr[i].is_string())
						{
							std::string res = arr[i].as_string();
							if (res.length() == 0)
								fprintf(stderr, "[\"\"]\n");
							else
								fprintf(stderr, "[%s]\n", res.c_str());
						}
						else if (arr[i].is_int())
						{
							fprintf(stderr, "[%d]\n", arr[i].as_int());
						}
						else if (arr[i].is_ulonglong())
						{
							fprintf(stderr, "[%llu]\n", arr[i].as_ulonglong());
						}
						else if (arr[i].is_float())
						{
							const void *ptr;
							size_t len;
							int data_type;
							arr[i].get_cell_nocopy(&ptr, &len, &data_type);
							size_t pos;
							for (pos = 0; pos < len; pos++)
								if (*((const char *)ptr + pos) == '.')
									break;
							if (pos != len)
								pos = len - pos - 1;
							else
								pos = 0;
							fprintf(stderr, "[%.*f]\n", (int)pos, arr[i].as_float());
						}
						else if (arr[i].is_double())
						{
							const void *ptr;
							size_t len;
							int data_type;
							arr[i].get_cell_nocopy(&ptr, &len, &data_type);
							size_t pos;
							for (pos = 0; pos < len; pos++)
								if (*((const char *)ptr + pos) == '.')
									break;
							if (pos != len)
								pos = len - pos - 1;
							else
								pos = 0;
							fprintf(stderr, "[%.*lf]\n", (int)pos, arr[i].as_double());
						}
						else if (arr[i].is_date())
						{
							fprintf(stderr, "[%s]\n", arr[i].as_string().c_str());
						}
						else if (arr[i].is_time())
						{
							fprintf(stderr, "[%s]\n", arr[i].as_string().c_str());
						}
						else if (arr[i].is_datetime())
						{
							fprintf(stderr, "[%s]\n", arr[i].as_string().c_str());
						}
						else if (arr[i].is_null())
						{
							fprintf(stderr, "[NULL]\n");
						}
						else
						{
							std::string res = arr[i].as_binary_string();
							if (res.length() == 0)
								fprintf(stderr, "[\"\"]\n");
							else
								fprintf(stderr, "[%s]\n", res.c_str());
						}
					}
					fprintf(stderr, "  __________ ROW END __________\n");
				}
			}
			else if (cursor.get_cursor_status() == MYSQL_STATUS_OK)
			{
				fprintf(stderr, "  OK. %llu ", cursor.get_affected_rows());
				if (cursor.get_affected_rows() == 1)
					fprintf(stderr, "row ");
				else
					fprintf(stderr, "rows ");
				fprintf(stderr, "affected. %d warnings. insert_id=%llu. %s\n",
						cursor.get_warnings(), cursor.get_insert_id(),
						cursor.get_info().c_str());
			}

			fprintf(stderr, "________________ RESULT SET END ________________\n\n");
		} while (cursor.next_result_set());

		if (resp->get_packet_type() == MYSQL_PACKET_ERROR)
		{
			fprintf(stderr, "ERROR. error_code=%d %s\n",
					task->get_resp()->get_error_code(),
					task->get_resp()->get_error_msg().c_str());
		}
		else if (resp->get_packet_type() == MYSQL_PACKET_OK) // just check origin APIs
		{
			fprintf(stderr, "OK. %llu ", task->get_resp()->get_affected_rows());
			if (task->get_resp()->get_affected_rows() == 1)
				fprintf(stderr, "row ");
			else
				fprintf(stderr, "rows ");
			fprintf(stderr, "affected. %d warnings. insert_id=%llu. %s\n",
					task->get_resp()->get_warnings(),
					task->get_resp()->get_last_insert_id(),
					task->get_resp()->get_info().c_str());
		}

		// ((Frpc::Reply*)task->user_data)->set_output("mysql ok");
		return;
	}
	void reply_collections(WFMySQLTask *task)
	{
		CHECK_MYSQL_STATUS(task);
		MySQLResponse *resp = task->get_resp();
		MySQLResultCursor cursor(resp);
		//const MySQLField *const *fields;
		std::vector<MySQLCell> arr;
		Frpc::CollectionArray ca;

		while (cursor.fetch_row(arr))
		{
			auto c = ca.add_collection();
			c->set_id(arr[0].as_int());
			c->set_name(arr[1].as_string());
			c->set_type_id(arr[2].as_int());
			c->set_about(arr[3].as_binary_string());
		}

		int size = ca.ByteSizeLong();
		char *buf = (char *)malloc(size);
		if (!buf)
		{
			((Frpc::Reply *)task->user_data)->set_bad_output(Frpc::Status::InternalServerError,"mysql malloc error");

			return;
		}
		if (ca.SerializeToArray(buf, size))
		{
			((Frpc::Reply *)task->user_data)->set_ok_output();
			((Frpc::Reply *)task->user_data)->add_output_nocopy(buf, size);
			return;
		}
		else
		{
			free(buf);
			((Frpc::Reply *)task->user_data)->set_bad_output(Frpc::Status::InternalServerError,"mysql pb serilization error");

			return;
		}
	}
	void reply_collectionitems(WFMySQLTask *task)
	{
	}
}